package com.gedi;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.RegexStringComparator;
import org.apache.hadoop.hbase.filter.RowFilter;
import org.apache.hadoop.hbase.mapreduce.TableInputFormat;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.mapreduce.TableOutputFormat;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;

import java.io.IOException;

import static com.sun.org.apache.xalan.internal.xsltc.compiler.util.Type.Text;

/**
 * @Author:RenPu
 * @Date: 2020/2/16 12:21
 * @Version: 1.0
 * @description:
 */
public class OrderApplication {

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {

        //1:创建job核心对象,所需参数为配置对象以及任务名称；
        Configuration conf = HBaseConfiguration.create();
        conf.set(HConstants.ZOOKEEPER_QUORUM,"hadoop:2181");
        Job job = Job.getInstance(conf,"order_compute");

        //2:设置输入数据格式以及输出数据格式
        job.setInputFormatClass(TableInputFormat.class);
        job.setOutputFormatClass(TableOutputFormat.class);

        //3：初始化mapper
        //创建扫描对象，设置过滤的规则
        Scan scan = new Scan();

        //2018年度账单统计
        //正则表达式过滤符合的结果：^.*:2018.*$
        //创建正则表达式的过滤对象,中需要的俩个参数，compareOp:代表设置比较的规则，大于，小于，或者等于等等，ByteArrayComparable:设置过滤的规则类型，正则，字符串等等方式
        //正则表达式的对象：RegexStringComparator；
        RowFilter rowFilter = new RowFilter(CompareFilter.CompareOp.EQUAL,new RegexStringComparator("^.*:2018.*$"));
        scan.setFilter(rowFilter);


        TableMapReduceUtil.initTableMapperJob(TableName.valueOf("t_order"),scan,OrderMapper.class,Text.class, DoubleWritable.class,job);

        //4：通过TableMapReduceUtils,工具类进行设置数据来源的表以及经过计算后的结果输出的表
        TableMapReduceUtil.initTableReducerJob("t_result",OrderReduce.class,job);

        //6：任务提交
        job.waitForCompletion(true);//输出打印日志



    }



}
